package com.allenkerr.rpc.client;

import com.allenkerr.rpc.common.RpcDecoder;
import com.allenkerr.rpc.common.RpcEncoder;
import com.allenkerr.rpc.common.RpcRequest;
import com.allenkerr.rpc.common.RpcResponse;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 框架的Rpc客户端，用于发送Rpc请求
 * 1. 在send()方法中实现netty client，向server发送请求
 * 2. 实现handler，接收response，设置给成员变量
 * <p>
 * Created by 小柯(Allen Kerr) on 17.5.23 20:16.
 */
public class RpcClient extends SimpleChannelInboundHandler<RpcResponse> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcClient.class);
    private final Object object = new Object();

    private String host;
    private int port;

    private RpcResponse response;

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public RpcResponse send(RpcRequest request) throws InterruptedException {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            //引导类
            Bootstrap bootstrap = new Bootstrap();
            //装配引导类
            bootstrap.group(eventLoopGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //向流水线中添加编码、解码、业务处理的handler
                            ChannelPipeline pipeline = socketChannel.pipeline()
                                    //编码，out
                                    .addLast(new RpcEncoder(RpcRequest.class))
                                    //解码，in 1
                                    .addLast(new RpcDecoder(RpcResponse.class))
                                    //解码，in 2
                                    .addLast(RpcClient.this);
                        }
                    }).option(ChannelOption.SO_KEEPALIVE, true);
            //连接到服务器
            ChannelFuture future = bootstrap.connect(host, port).sync();
            future.channel().writeAndFlush(request).sync();

            //用线程等待的方式决定是否关闭连接
            //功能：先在此阻塞，等待获取到服务端的回调后，被唤醒，因此网络连接不会被关闭
            synchronized (object) {
                object.wait();
            }
            if (response != null) {
                //服务器等待通道关闭，因为使用sync()，所以关闭操作会被阻塞
                future.channel().closeFuture().sync();
            }
            return response;
        } finally {
            //所有工作完成，关闭线程池
//            eventLoopGroup.shutdownGracefully().sync();
            //关闭线程池
            eventLoopGroup.shutdownGracefully();
        }
    }

    /**
     * 读取服务端返回的结果
     *
     * @param ctx
     * @param response
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response) throws Exception {
        this.response = response;
        synchronized (object) {
            object.notifyAll();
        }
    }

    /**
     * 异常处理
     *
     * @param ctx
     * @param cause
     * @throws Exception
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.error("Rpc Client Caught Exception: ", cause);
        ctx.close();
    }
}
